package akka.mapreduce.actors

import scala.collection.mutable.HashMap

import akka.actor._
import akka.mapreduce.ReduceData
import akka.mapreduce.Result


class AggregateActor extends Actor {
  val finalReduceMap = new HashMap[String, Int]

  def receive = {
    case Result => sender ! finalReduceMap.toString()
    case ReduceData(dataList) => aggregateInMemory(dataList)
  }

  def aggregateInMemory(reduceDataMap: Map[String, Int]): Unit = {
    for((key, value) <- reduceDataMap){
      if(finalReduceMap contains key)
        finalReduceMap(key) = ( value + finalReduceMap.get(key).get)
      else
        finalReduceMap += (key -> value)
    }
  
 }
}
